package zju.lzy.service.message;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import zju.lzy.dto.IOTMessage;
import zju.lzy.dto.MessageReq;
import zju.lzy.mapper.MessageMapper;
import zju.lzy.service.IOTProperties;
import zju.lzy.util.BSUtil;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;


/**
 * @author lzy
 */
@Component
public class MessageReceiver {


    private static final Logger logger = LoggerFactory.getLogger(MessageReceiver.class);

    Deque<IOTMessage> cacheQueue = new ConcurrentLinkedDeque<>();

    HashSet<String> topicSet = new HashSet<>();

    CountDownLatch latch = new CountDownLatch(0);

    private MqttClient mqttClient;

    @Autowired
    private MessageMapper messageMapper;

    @Autowired
    private IOTProperties iotProperties;


    public void start() throws MqttException {
        mqttClient = new MqttClient("tcp://localhost:1883", "receiver", new MemoryPersistence());
        mqttClient.connect();
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                logger.info("连接成功，当前Server是" + serverURI);
            }

            @Override
            public void connectionLost(Throwable cause) {
                cause.printStackTrace();
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                IOTMessage msg = JSONObject.parseObject(
                        new String(message.getPayload()),
                        IOTMessage.class
                );
                logger.info(
                        "接到消息，topic是：{}",
                        msg.getTopic()
                );
                // 存入缓存
                cacheQueue.push(msg);
                if (cacheQueue.size() > 10000) {
                    messageMapper.batchStoreMessage(new ArrayList<>(cacheQueue));
                    cacheQueue.clear();
                    logger.info("缓存量过大，强制写入");
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {

            }
        });

        BSUtil.tpe.execute(() -> {
            while (true) {
                try {
                    Thread.sleep(iotProperties.getCacheFlushInterval() * 1000L);
                    if (cacheQueue.size() != 0) {
                        messageMapper.batchStoreMessage(new ArrayList<>(cacheQueue));
                        logger.info("进行一次缓存flush，数据量为：{}", cacheQueue.size());
                        cacheQueue.clear();
                    } else {
                        logger.info("当前缓存为空，无需写入负载");
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

    }


    public MessageReceiver() {
    }

    /**
     * 可以多次注册调用逻辑,
     *
     * @param topic
     */
    public void subscribeTopic(String topic) {
        if (!topicSet.contains(topic)) {
            try {
                mqttClient.subscribe(topic, 2);
            } catch (MqttException e) {
                e.printStackTrace();
            } finally {
                topicSet.add(topic);
            }
        }
    }

    public Set<String> getSubscribedTopics() {
        return topicSet;
    }

    /**
     * 根据查询条件，对缓存中的数据进行查询
     *
     * @param messageReq
     * @return
     */
    public List<IOTMessage> queryCachedMessages(MessageReq messageReq) {
        return cacheQueue.stream()
                .filter(messageReq.generateFilter()::isWanted)
                .collect(Collectors.toList());
    }

    public List<IOTMessage> getDataBySender(Integer sender) {
        return cacheQueue.stream()
                .filter(data -> data.getSender().equals(sender))
                .collect(Collectors.toList());
    }
}
